package org.pentaho.di.trans.steps.redis;

import org.pentaho.di.core.Const;
import org.pentaho.di.core.Counter;
import org.pentaho.di.core.database.DatabaseMeta;
import org.pentaho.di.core.exception.KettleException;
import org.pentaho.di.core.exception.KettleXMLException;
import org.pentaho.di.core.xml.XMLHandler;
import org.pentaho.di.i18n.BaseMessages;
import org.pentaho.di.repository.ObjectId;
import org.pentaho.di.repository.Repository;
import org.pentaho.di.trans.step.BaseStepMeta;
import org.w3c.dom.Node;
import redis.clients.jedis.Jedis;
import redis.clients.jedis.JedisPool;
import redis.clients.jedis.JedisPoolConfig;
import redis.clients.jedis.JedisSentinelPool;
import redis.clients.util.Pool;

import java.util.ArrayList;
import java.util.HashSet;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.Set;

public abstract class RedisBaseStepMeta extends BaseStepMeta {
    private static Class<?> PKG = RedisBaseStepMeta.class; // for i18n purposes, needed by Translator2!! $NON-NLS-1$

    public static final Integer SERVERT_TYPE_STANDALONE = 1;
    public static final Integer SERVERT_TYPE_SENTINEL = 2;
    public static final Integer SERVERT_TYPE_CLUSTER = 3;

    private Integer serverType;
    private String masterName;
    private List<RedisServer> servers;

    private Integer maxTotal;
    private Integer maxIdle;
    private Integer maxWaitMillis;
    private String testOnBorrow;
    private String testOnReturn;
    private Integer timeout;

    public RedisBaseStepMeta() {
        super();
        setDefault();
    }

    public Pool<Jedis> createJedisPool() throws KettleException {
        if ((null == servers) || (servers.size() == 0)) {
            throw new KettleException("No redis server has been configured!");
        }

        Pool<Jedis> pool = null;
        JedisPoolConfig jedisPoolConfig = new JedisPoolConfig();
        jedisPoolConfig.setMaxTotal(maxTotal);
        jedisPoolConfig.setMaxIdle(maxIdle);
        jedisPoolConfig.setMaxWaitMillis(maxWaitMillis);
        jedisPoolConfig.setTestOnBorrow("Y".equals(testOnBorrow));//jedis 第一次启动时，会报错
        jedisPoolConfig.setTestOnReturn("Y".equals(testOnReturn));
        if (SERVERT_TYPE_STANDALONE == serverType) {
            // 初始化JedisPool
            RedisServer redisServer = servers.get(0);
            pool = new JedisPool(jedisPoolConfig, redisServer.getHostname(), Integer.valueOf(redisServer.getPort()), timeout);
        } else if (SERVERT_TYPE_SENTINEL == serverType) {
            String password =null;
            Iterator<RedisBaseStepMeta.RedisServer> it = servers.iterator();
            Set<String> sentinels = new HashSet<String>();
            while (it.hasNext()) {
                RedisBaseStepMeta.RedisServer hostAndPort= it.next();
                password =hostAndPort.getAuth();
                sentinels.add(hostAndPort.getHostname()+":"+hostAndPort.getPort());
            }
            pool = new JedisSentinelPool(masterName, sentinels, jedisPoolConfig, timeout, password);
        } else {
            throw new KettleException("Unsupported pool type:" + this.serverType);
        }

        return pool;
    }

    protected void cloneBase(RedisBaseStepMeta retval) {
        retval.setMasterName(this.masterName);
        retval.setServers(this.servers);
        retval.setServerType(this.serverType);
    }

    private void allocate() {
        servers = new ArrayList<>();
    }

    public void setDefault() {
        this.serverType = SERVERT_TYPE_STANDALONE;
        this.masterName = null;
        maxTotal = 1000;
        maxIdle = 100;
        maxWaitMillis = 500;
        testOnBorrow = "Y";
        testOnReturn = "Y";
        timeout = 1000;
        allocate();
    }

    protected String getXMLPart() {
        StringBuilder retval = new StringBuilder();
        retval.append("    " + XMLHandler.addTagValue("servertype", this.getServerType()));
        retval.append("    " + XMLHandler.addTagValue("mastername", this.getMasterName()));

        retval.append("    " + XMLHandler.addTagValue("maxtotal", this.getMaxTotal()));
        retval.append("    " + XMLHandler.addTagValue("maxidle", this.getMaxIdle()));
        retval.append("    " + XMLHandler.addTagValue("maxwaitmillis", this.getMaxWaitMillis()));
        retval.append("    " + XMLHandler.addTagValue("testonborrow", this.getTestOnBorrow()));
        retval.append("    " + XMLHandler.addTagValue("testonreturn", this.getTestOnReturn()));
        retval.append("    " + XMLHandler.addTagValue("timeout", this.getTimeout()));

        retval.append("    <servers>").append(Const.CR);
        List<RedisServer> servers = this.getServers();
        if (servers != null) {
            Iterator<RedisServer> iterator = servers.iterator();
            while (iterator.hasNext()) {
                RedisServer addr=iterator.next();
                retval.append("      <server>").append(Const.CR);
                retval.append("        ").append(XMLHandler.addTagValue("hostname", addr.getHostname()));
                retval.append("        ").append(XMLHandler.addTagValue("port", addr.getPort()));
                retval.append("        ").append(XMLHandler.addTagValue("auth", addr.getAuth()));
                retval.append("      </server>").append(Const.CR);
            }
        }
        retval.append("    </servers>").append(Const.CR);

        return retval.toString();
    }

    protected void readData(Node stepnode) throws KettleXMLException {
        try {
            this.serverType = Integer.valueOf(XMLHandler.getTagValue(stepnode, "servertype"));
            this.masterName = XMLHandler.getTagValue(stepnode, "mastername");
            Node serverNodes = XMLHandler.getSubNode(stepnode, "servers");
            int nrservers = XMLHandler.countNodes(serverNodes, "server");
            allocate();

            for (int i = 0; i < nrservers; i++) {
                Node fnode = XMLHandler.getSubNodeByNr(serverNodes, "server", i);
                RedisServer redisServer = new RedisServer();
                redisServer.setHostname(XMLHandler.getTagValue(fnode, "hostname"));
                redisServer.setPort(XMLHandler.getTagValue(fnode, "port"));
                redisServer.setAuth(XMLHandler.getTagValue(fnode, "auth"));

                servers.add(redisServer);
            }
            maxTotal = Integer.valueOf(XMLHandler.getTagValue(stepnode, "maxtotal"));
            maxIdle = Integer.valueOf(XMLHandler.getTagValue(stepnode, "maxidle"));
            maxWaitMillis = Integer.valueOf(XMLHandler.getTagValue(stepnode, "maxwaitmillis"));
            testOnBorrow = XMLHandler.getTagValue(stepnode, "testonborrow");
            testOnReturn = XMLHandler.getTagValue(stepnode, "testonreturn");
            timeout = Integer.valueOf(XMLHandler.getTagValue(stepnode, "timeout"));
        } catch (Exception e) {
            throw new KettleXMLException(BaseMessages.getString(PKG, "RedisInputMeta.Exception.UnableToReadStepInfo"),
                    e);
        }
    }

    public void readRep(Repository rep, ObjectId id_step, List<DatabaseMeta> databases, Map<String, Counter> counters)
            throws KettleException {
        try {
            this.serverType = (int)rep.getStepAttributeInteger(id_step, "servertype");
            this.masterName = rep.getStepAttributeString(id_step, "mastername");
            int nrservers = rep.countNrStepAttributes(id_step, "servercount");
            allocate();

            for (int i = 0; i < nrservers; i++) {
                RedisServer nrserversmap=new RedisServer();
                nrserversmap.setHostname(rep.getStepAttributeString(id_step, i, "hostname"));
                nrserversmap.setPort(rep.getStepAttributeString(id_step, i, "port"));
                nrserversmap.setAuth(rep.getStepAttributeString(id_step, i, "auth"));
                servers.add(nrserversmap);
            }

            maxTotal = (int)rep.getStepAttributeInteger(id_step, "maxtotal");
            maxIdle = (int)rep.getStepAttributeInteger(id_step, "maxidle");
            maxWaitMillis = (int)rep.getStepAttributeInteger(id_step, "maxwaitmillis");
            testOnBorrow = rep.getStepAttributeString(id_step, "testonborrow");
            testOnReturn = rep.getStepAttributeString(id_step, "testonreturn");
            timeout = (int)rep.getStepAttributeInteger(id_step, "timeout");

        } catch (Exception e) {
            throw new KettleException(BaseMessages.getString(PKG,
                    "RedisInputMeta.Exception.UnexpectedErrorReadingStepInfo"), e);
        }
    }

    public void saveRep(Repository rep, ObjectId id_transformation, ObjectId id_step)
            throws KettleException {
        try {
            rep.saveStepAttribute(id_transformation, id_step, "servertype", this.serverType);
            rep.saveStepAttribute(id_transformation, id_step, "mastername", this.masterName);
            int i = 0;
            List<RedisServer> servers = this.getServers();
            if (servers != null) {
                rep.saveStepAttribute(id_transformation, id_step, "servercount", servers.size());
                Iterator<RedisServer> iterator = servers.iterator();
                while (iterator.hasNext()) {
                    RedisServer addr=iterator.next();
                    rep.saveStepAttribute(id_transformation, id_step, i, "hostname", addr.getHostname());
                    rep.saveStepAttribute(id_transformation, id_step, i, "port", addr.getPort());
                    rep.saveStepAttribute(id_transformation, id_step, i, "auth", addr.getAuth());
                    i++;
                }
            } else {
                rep.saveStepAttribute(id_transformation, id_step, "servercount", 1);
            }
            rep.saveStepAttribute(id_transformation, id_step, "maxtotal", maxTotal);
            rep.saveStepAttribute(id_transformation, id_step, "maxidle", maxIdle);
            rep.saveStepAttribute(id_transformation, id_step, "maxwaitmillis", maxWaitMillis);
            rep.saveStepAttribute(id_transformation, id_step, "testonborrow", testOnBorrow );
            rep.saveStepAttribute(id_transformation, id_step, "testonreturn", testOnReturn );
            rep.saveStepAttribute(id_transformation, id_step, "timeout", timeout);
        } catch (Exception e) {
            throw new KettleException(BaseMessages.getString(PKG,
                    "RedisInputMeta.Exception.UnexpectedErrorSavingStepInfo"), e);
        }
    }

    public String getMasterName() {
        return masterName;
    }

    public void setMasterName(String masterName) {
        this.masterName = masterName;
    }

    public List<RedisServer> getServers() {
        return servers;
    }

    public void setServers(List<RedisServer> servers) {
        this.servers = servers;
    }

    public Integer getServerType() {
        return serverType;
    }

    public void setServerType(Integer serverType) {
        this.serverType = serverType;
    }

    public Integer getMaxTotal() {
        return maxTotal;
    }

    public void setMaxTotal(Integer maxTotal) {
        this.maxTotal = maxTotal;
    }

    public Integer getMaxIdle() {
        return maxIdle;
    }

    public void setMaxIdle(Integer maxIdle) {
        this.maxIdle = maxIdle;
    }

    public Integer getMaxWaitMillis() {
        return maxWaitMillis;
    }

    public void setMaxWaitMillis(Integer maxWaitMillis) {
        this.maxWaitMillis = maxWaitMillis;
    }

    public String getTestOnBorrow() {
        return testOnBorrow;
    }

    public void setTestOnBorrow(String testOnBorrow) {
        this.testOnBorrow = testOnBorrow;
    }

    public String getTestOnReturn() {
        return testOnReturn;
    }

    public void setTestOnReturn(String testOnReturn) {
        this.testOnReturn = testOnReturn;
    }

    public Integer getTimeout() {
        return timeout;
    }

    public void setTimeout(Integer timeout) {
        this.timeout = timeout;
    }

    public static class RedisServer {
        private String hostname;
        private String port = "6379";
        private String auth;

        public String getHostname() {
            return (null == hostname) ? "" : hostname;
        }

        public void setHostname(String hostname) {
            this.hostname = hostname;
        }

        public String getPort() {
            return (null == port) ? "" : port;
        }

        public void setPort(String port) {
            this.port = port;
        }

        public String getAuth() {
            return (auth == null) ? "" : auth;
        }

        public void setAuth(String auth) {
            this.auth = auth;
        }
    }
}
